﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using RabbitMQ.Client;
using System.Threading.Tasks;
using System.Threading;
using RabbitMQ.Client.Events;
using System.Diagnostics;

namespace RMQLab
{
    class Program
    {
        static void Main(string[] args)
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            Send();
            sw.Stop();
            Console.WriteLine(sw.ElapsedMilliseconds);
            Console.ReadLine();
            sw.Reset();
            sw.Start();
            Receive();
            Console.WriteLine(sw.ElapsedMilliseconds);
            sw.Stop();
            //NewTask(args);
            //Task.Factory.StartNew(Worker);
            //Task.Factory.StartNew(Worker);
            //Task.Factory.StartNew(Send);
            //Task.Factory.StartNew(Receive);
            //Task.Factory.StartNew(Receive);

            Console.ReadLine();

        }
        private static string GetMessage(string[] args)
        {
            return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
        }

        static void Send()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                connection.ConnectionBlocked += new ConnectionBlockedEventHandler((c, e) =>
                {
                    Console.WriteLine("blocked");
                });
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    string message = "msg";
                    for (int i = 0; i < 20000; i++)
                    //int i = 0;
                    //while(true)
                    {                        
                        var body = Encoding.UTF8.GetBytes(message + i);
                        
                        channel.BasicPublish(string.Empty, "hello", null, body);
                        
                        if (i%2000==0)
                        {
                            var ok = channel.QueueDeclare("hello", false, false, false, null);
                            Console.WriteLine(" [x] Sent {0}", message + i);
                            Console.WriteLine("queue length:{0}", ok.MessageCount); 
                        }
                        //i++;
                    }

                }
            }
        }

        static void Receive()
        {
            int threadId = Thread.CurrentThread.ManagedThreadId;
            var factory = new ConnectionFactory() { HostName = "localhost" };
            int counter = 0;
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume("hello", true, consumer);
                    while (true)
                    {
                        var eventArgs = consumer.Queue.Dequeue();
                        byte[] body = eventArgs.Body;
                        string message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0} in thread:{1}", message, threadId);
                        //if (++counter>=1000000)
                        //{
                        //    break;
                        //}
                    }

                }
            }
        }

        static void NewTask(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("task_queue", true, false, false, null);

                    var message = GetMessage(args);
                    var body = Encoding.UTF8.GetBytes(message);
                    
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);

                    channel.BasicPublish("", "task_queue", properties, body);
                    Console.WriteLine(" [x] Sent {0}", message);
                }
            }
        }

        static void Worker()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("task_queue", true, false, false, null);

                    channel.BasicQos(0, 1, false);
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume("task_queue", false, consumer);

                    Console.WriteLine(" [*] Waiting for messages. " +
                                      "To exit press CTRL+C");
                    while (true)
                    {
                        BasicDeliverEventArgs ea = consumer.Queue.Dequeue();

                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0}", message);

                        int dots = message.Split('.').Length - 1;
                        Thread.Sleep(dots * 1000);

                        Console.WriteLine(" [x] Done");

                        channel.BasicAck(ea.DeliveryTag, false);
                    }
                }
            }


        }
    }
}
